package cn.doitedu;

import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.apache.hadoop.io.BytesWritable;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/4/24
 * @Desc: 学大数据，上多易教育
 *
 *    把一组bitmap聚合成一个bitmap
 **/
public class AggBitmapUDAF extends UDAF {
    public static class AggBitmapUDAFEvaluator implements UDAFEvaluator {

        Acc acc ;

        public AggBitmapUDAFEvaluator(){
            init();
        }

        @Override
        public void init() {
            try {
                this.acc = new Acc();

            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }



        public boolean iterate(BytesWritable bitmapBytesWritable) throws IOException {
            byte[] bitmapBytes = bitmapBytesWritable.getBytes();
            // 反序列化收到的一个bitmap
            Roaring64Bitmap bitmap = BitmapUtils.de(bitmapBytes);

            // 取出自己的累加器中的bitmap
            Roaring64Bitmap selfBitmap = acc.getBitmap();


            // 合并两个bitmap
            selfBitmap.or(bitmap);

            // 将合并好的bitmap序列化，并设置到自己的累加器中
            acc.setBitmap(selfBitmap);

            return true;
        }


        public Acc terminatePartial() {
            return this.acc;
        }


        public boolean merge(Acc otherAcc) throws IOException {
            // 取出输入的累加器中的bitmap对象
            Roaring64Bitmap bitmap = otherAcc.getBitmap();
            // 取出自己的累加器中的bitmap对象
            Roaring64Bitmap selfBitmap = acc.getBitmap();

            // 合并两个bitmap
            selfBitmap.or(bitmap);

            // 将合并好的bitmap更新到自己的累加器中
            acc.setBitmap(selfBitmap);

            return true;

        }


        public byte[] terminate() {

            return this.acc.getBitmapBytes();

        }




    }








}
